package drds.binlog.parse.inbound.mysql.dbsync;

import com.google.protobuf.ByteString;
import drds.binlog.LogEvent;
import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.event.*;
import drds.binlog.event.TableMapLogEvent.ColumnInfo;
import drds.binlog.event.mariadb.AnnotateRowsEvent;
import drds.binlog.filter.aviater.AviaterRegexFilter;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.parse.exception.TableIdNotFoundException;
import drds.binlog.parse.inbound.BinlogParser;
import drds.binlog.parse.inbound.FieldMetaData;
import drds.binlog.parse.inbound.TableMetaData;
import drds.binlog.parse.inbound.mysql.ddl.DdlResult;
import drds.binlog.parse.inbound.mysql.ddl.DruidDdlParser;
import drds.binlog.parse.inbound.mysql.ddl.SimpleDdlParser;
import drds.binlog.protocol.entry_protocol_proto.*;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.charset.Charset;
import java.sql.Types;
import java.util.Arrays;
import java.util.BitSet;
import java.util.List;

/**
 * 基于{@linkplain LogEvent}转化为Entry对象的处理
 */
public class LogEventConvert extends AbstractLifeCycle implements BinlogParser<LogEvent>
{

    public static final String XA_XID = "XA_XID";
    public static final String XA_TYPE = "XA_TYPE";
    public static final String XA_START = "XA START";
    public static final String XA_END = "XA END";
    public static final String XA_COMMIT = "XA COMMIT";
    public static final String XA_ROLLBACK = "XA ROLLBACK";
    public static final String ISO_8859_1 = "ISO-8859-1";
    public static final String UTF_8 = "UTF-8";
    public static final int TINYINT_MAX_VALUE = 256;
    public static final int SMALLINT_MAX_VALUE = 65536;
    public static final int MEDIUMINT_MAX_VALUE = 16777216;
    public static final long INTEGER_MAX_VALUE = 4294967296L;
    public static final BigInteger BIGINT_MAX_VALUE = new BigInteger("18446744073709551616");
    public static final int version = 1;
    public static final String BEGIN = "BEGIN";
    public static final String COMMIT = "COMMIT";
    public static final Logger logger = LoggerFactory.getLogger(LogEventConvert.class);

    private volatile AviaterRegexFilter nameFilter;                                                          // 运行时引用可能会有变化，比如规则发生变化时
    private volatile AviaterRegexFilter nameBlackFilter;

    private TableMetaDataCache tableMetaDataCache;
    private Charset charset = Charset.defaultCharset();
    private boolean filterQueryDcl = false;
    private boolean filterQueryDml = false;
    private boolean filterQueryDdl = false;
    // 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92
    private boolean filterTableError = false;
    // 新增rows过滤，用于仅订阅除rows以外的数据
    private boolean filterRows = false;
    private boolean useDruidDdlFilter = true;

    public LogEventConvert()
    {

    }

    public static TransactionBegin createTransactionBegin(long threadId)
    {
        TransactionBegin.Builder beginBuilder = TransactionBegin.newBuilder();
        beginBuilder.setThreadId(threadId);
        return beginBuilder.build();
    }

    public static TransactionEnd createTransactionEnd(long transactionId)
    {
        TransactionEnd.Builder endBuilder = TransactionEnd.newBuilder();
        endBuilder.setTransactionId(String.valueOf(transactionId));
        return endBuilder.build();
    }

    public static Pair createSpecialPair(String key, String value)
    {
        Pair.Builder pairBuilder = Pair.newBuilder();
        pairBuilder.setKey(key);
        pairBuilder.setValue(value);
        return pairBuilder.build();
    }

    public static Entry createEntry(Header header, EntryType entryType, ByteString storeValue)
    {
        Entry.Builder entryBuilder = Entry.newBuilder();
        entryBuilder.setHeader(header);
        entryBuilder.setEntryType(entryType);
        entryBuilder.setStoreValue(storeValue);
        return entryBuilder.build();
    }

    @Override
    public Entry parse(LogEvent logEvent, boolean isSeek) throws BinlogParseException
    {
        if (logEvent == null || logEvent instanceof UnknownLogEvent)
        {
            return null;
        }

        int eventType = logEvent.getLogHeader().getType();
        switch (eventType)
        {
            case LogEvent.QUERY_EVENT:
                return parseQueryEvent((QueryLogEvent) logEvent, isSeek);
            case LogEvent.XID_EVENT:
                return parseXidEvent((XidLogEvent) logEvent);
            case LogEvent.TABLE_MAP_EVENT:
                break;
            case LogEvent.WRITE_ROWS_EVENT_V1:
            case LogEvent.WRITE_ROWS_EVENT:
                return parseRowsEvent((WriteRowsLogEvent) logEvent);
            case LogEvent.UPDATE_ROWS_EVENT_V1:
            case LogEvent.UPDATE_ROWS_EVENT:
                return parseRowsEvent((UpdateRowsLogEvent) logEvent);
            case LogEvent.DELETE_ROWS_EVENT_V1:
            case LogEvent.DELETE_ROWS_EVENT:
                return parseRowsEvent((DeleteRowsLogEvent) logEvent);
            case LogEvent.ROWS_QUERY_LOG_EVENT:
                return parseRowsQueryEvent((RowsQueryLogEvent) logEvent);
            case LogEvent.ANNOTATE_ROWS_EVENT:
                return parseAnnotateRowsEvent((AnnotateRowsEvent) logEvent);
            case LogEvent.USER_VAR_EVENT:
                return parseUserVarLogEvent((UserVarLogEvent) logEvent);
            case LogEvent.INTVAR_EVENT:
                return parseIntrvarLogEvent((IntvarLogEvent) logEvent);
            case LogEvent.RAND_EVENT:
                return parseRandLogEvent((RandLogEvent) logEvent);
            case LogEvent.GTID_LOG_EVENT:
                return parseGTIDLogEvent((GtidLogEvent) logEvent);
            case LogEvent.HEARTBEAT_LOG_EVENT:
                return parseHeartbeatLogEvent();
            default:
                break;
        }

        return null;
    }

    public void reset()
    {
        // do nothing
        if (tableMetaDataCache != null)
        {
            tableMetaDataCache.clearTableMeta();
        }
    }

    private Entry parseHeartbeatLogEvent()
    {
        Header.Builder headerBuilder = Header.newBuilder();
        headerBuilder.setEventType(EventType.MHEARTBEAT);
        Entry.Builder entryBuilder = Entry.newBuilder();
        entryBuilder.setHeader(headerBuilder.build());
        entryBuilder.setEntryType(EntryType.HEARTBEAT);
        return entryBuilder.build();
    }

    private Entry parseGTIDLogEvent(GtidLogEvent gtidLogEvent)
    {
        LogHeader logHeader = gtidLogEvent.getLogHeader();
        Pair.Builder builder = Pair.newBuilder();
        builder.setKey("gtid");
        builder.setValue(gtidLogEvent.getGtidString());

        if (gtidLogEvent.getLastCommitted() != null)
        {
            builder.setKey("lastCommitted");
            builder.setValue(String.valueOf(gtidLogEvent.getLastCommitted()));
            builder.setKey("sequenceNumber");
            builder.setValue(String.valueOf(gtidLogEvent.getSequenceNumber()));
        }

        Header header = createHeader(logHeader, "", "", EventType.GTID);
        return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());
    }

    private Entry parseQueryEvent(QueryLogEvent queryLogEvent, boolean isSeek)
    {
        String queryString = queryLogEvent.getQuery();
        if (StringUtils.startsWithIgnoreCase(queryString, XA_START))
        {
            // xa start use TransactionBegin
            TransactionBegin.Builder beginBuilder = TransactionBegin.newBuilder();
            beginBuilder.setThreadId(queryLogEvent.getSessionId());
            beginBuilder.addProps(createSpecialPair(XA_TYPE, XA_START));
            beginBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_START)));
            TransactionBegin transactionBegin = beginBuilder.build();
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_END))
        {
            // xa start use TransactionEnd
            TransactionEnd.Builder endBuilder = TransactionEnd.newBuilder();
            endBuilder.setTransactionId(String.valueOf(0L));
            endBuilder.addProps(createSpecialPair(XA_TYPE, XA_END));
            endBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_END)));
            TransactionEnd transactionEnd = endBuilder.build();
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_COMMIT))
        {
            // xa commit
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", EventType.XACOMMIT);
            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            rowChangeBuider.setSql(queryString);
            rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_COMMIT));
            rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_COMMIT)));
            rowChangeBuider.setEventType(EventType.XACOMMIT);
            return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
        } else if (StringUtils.startsWithIgnoreCase(queryString, XA_ROLLBACK))
        {
            // xa rollback
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", EventType.XAROLLBACK);
            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            rowChangeBuider.setSql(queryString);
            rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_ROLLBACK));
            rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_ROLLBACK)));
            rowChangeBuider.setEventType(EventType.XAROLLBACK);
            return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
        } else if (StringUtils.endsWithIgnoreCase(queryString, BEGIN))
        {
            TransactionBegin transactionBegin = createTransactionBegin(queryLogEvent.getSessionId());
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
        } else if (StringUtils.endsWithIgnoreCase(queryString, COMMIT))
        {
            TransactionEnd transactionEnd = createTransactionEnd(0L); // MyISAM可能不会有xid事件
            Header header = createHeader(queryLogEvent.getLogHeader(), "", "", null);
            return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
        } else
        {
            boolean notFilter = false;
            EventType eventType = EventType.QUERY;
            String tableName = null;
            String schemaName = null;
            if (useDruidDdlFilter)
            {
                List<DdlResult> ddlResultList = DruidDdlParser.parse(queryString, queryLogEvent.getDbName());
                for (DdlResult result : ddlResultList)
                {
                    if (!processFilter(queryString, result))
                    {
                        // 只要有一个数据不进行过滤
                        notFilter = true;
                    }
                }
                if (ddlResultList.size() > 0)
                {
                    // 如果针对多行的DDL,只能取第一条
                    eventType = ddlResultList.get(0).getType();
                    schemaName = ddlResultList.get(0).getSchemaName();
                    tableName = ddlResultList.get(0).getTableName();
                }
            } else
            {
                DdlResult result = SimpleDdlParser.parse(queryString, queryLogEvent.getDbName());
                if (!processFilter(queryString, result))
                {
                    notFilter = true;
                }

                eventType = result.getType();
                schemaName = result.getSchemaName();
                tableName = result.getTableName();
            }

            if (!notFilter)
            {
                // 如果是过滤的数据就不处理了
                return null;
            }

            if (!isSeek)
            {
                // 使用新的表结构元数据管理方式
                EntryPosition position = createPosition(queryLogEvent.getLogHeader());
                tableMetaDataCache.apply(position, queryLogEvent.getDbName(), queryString, null);
            }

            Header header = createHeader(queryLogEvent.getLogHeader(), schemaName, tableName, eventType);
            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            if (eventType != EventType.QUERY && eventType != EventType.INSERT && eventType != EventType.UPDATE
                    && eventType != EventType.DELETE)
            {
                rowChangeBuider.setIsDdl(true);
            }
            rowChangeBuider.setSql(queryString);
            if (StringUtils.isNotEmpty(queryLogEvent.getDbName()))
            {// 可能为空
                rowChangeBuider.setDdlSchemaName(queryLogEvent.getDbName());
            }
            rowChangeBuider.setEventType(eventType);
            return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
        }
    }

    private String getXaXid(String queryString, String type)
    {
        return StringUtils.substringAfter(queryString, type);
    }

    private boolean processFilter(String queryString, DdlResult result)
    {
        String schemaName = result.getSchemaName();
        String tableName = result.getTableName();
        // fixed issue https://github.com/alibaba/canal/issues/58
        // 更新下table metadata cache
        if (tableMetaDataCache != null
                && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME))
        {
            // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
            for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult())
            {
                String schemaName0 = renameResult.getSchemaName();
                String tableName0 = renameResult.getTableName();
                if (StringUtils.isNotEmpty(tableName0))
                {
                    // 如果解析到了正确的表信息，则根据全名进行清除
                    tableMetaDataCache.clearTableMeta(schemaName0, tableName0);
                } else
                {
                    // 如果无法解析正确的表信息，则根据schema进行清除
                    tableMetaDataCache.clearTableMetaWithSchemaName(schemaName0);
                }
            }
        }

        // fixed issue https://github.com/alibaba/canal/issues/58
        if (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE
                || result.getType() == EventType.CREATE || result.getType() == EventType.TRUNCATE
                || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX
                || result.getType() == EventType.DINDEX)
        { // 针对DDL类型

            if (filterQueryDdl)
            {
                return true;
            }

            if (StringUtils.isEmpty(tableName)
                    || (result.getType() == EventType.RENAME && StringUtils.isEmpty(result.getOriTableName())))
            {
                // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析
                throw new BinlogParseException("SimpleDdlParser process query failed. pls submit issue with this queryString: "
                        + queryString + " , and DdlResult: " + result.toString());
                // return null;
            } else
            {
                // check name eventFilter
                String name = schemaName + "." + tableName;
                if (nameFilter != null && !nameFilter.filter(name))
                {
                    if (result.getType() == EventType.RENAME)
                    {
                        // rename校验只要源和目标满足一个就进行操作
                        if (nameFilter != null
                                && !nameFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName()))
                        {
                            return true;
                        }
                    } else
                    {
                        // 其他情况返回null
                        return true;
                    }
                }

                if (nameBlackFilter != null && nameBlackFilter.filter(name))
                {
                    if (result.getType() == EventType.RENAME)
                    {
                        // rename校验只要源和目标满足一个就进行操作
                        if (nameBlackFilter != null
                                && nameBlackFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName()))
                        {
                            return true;
                        }
                    } else
                    {
                        // 其他情况返回null
                        return true;
                    }
                }
            }
        } else if (result.getType() == EventType.INSERT || result.getType() == EventType.UPDATE
                || result.getType() == EventType.DELETE)
        {
            // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
            if (filterQueryDml)
            {
                return true;
            }
        } else if (filterQueryDcl)
        {
            return true;
        }

        return false;
    }

    private Entry parseRowsQueryEvent(RowsQueryLogEvent rowsQueryLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }
        // mysql5.6支持，需要设置binlog-rows-query-log-events=1，可详细打印原始DML语句
        String queryString = null;
        try
        {
            queryString = new String(rowsQueryLogEvent.getRowsQuery().getBytes(ISO_8859_1), charset.name());
            String tableName = null;
            if (useDruidDdlFilter)
            {
                List<DdlResult> results = DruidDdlParser.parse(queryString, null);
                if (results.size() > 0)
                {
                    tableName = results.get(0).getTableName();
                }
            }

            return buildQueryEntry(queryString, rowsQueryLogEvent.getLogHeader(), tableName);
        } catch (UnsupportedEncodingException e)
        {
            throw new BinlogParseException(e);
        }
    }

    private Entry parseAnnotateRowsEvent(AnnotateRowsEvent annotateRowsEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }
        // mariaDb支持，需要设置binlog_annotate_row_events=true，可详细打印原始DML语句
        String queryString = null;
        try
        {
            queryString = new String(annotateRowsEvent.getRowsQuery().getBytes(ISO_8859_1), charset.name());
            return buildQueryEntry(queryString, annotateRowsEvent.getLogHeader());
        } catch (UnsupportedEncodingException e)
        {
            throw new BinlogParseException(e);
        }
    }

    private Entry parseUserVarLogEvent(UserVarLogEvent userVarLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }

        return buildQueryEntry(userVarLogEvent.getQuery(), userVarLogEvent.getLogHeader());
    }

    private Entry parseIntrvarLogEvent(IntvarLogEvent intvarLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }

        return buildQueryEntry(intvarLogEvent.getQuery(), intvarLogEvent.getLogHeader());
    }

    private Entry parseRandLogEvent(RandLogEvent randLogEvent)
    {
        if (filterQueryDml)
        {
            return null;
        }

        return buildQueryEntry(randLogEvent.getQuery(), randLogEvent.getLogHeader());
    }

    private Entry parseXidEvent(XidLogEvent xidLogEvent)
    {
        TransactionEnd transactionEnd = createTransactionEnd(xidLogEvent.getXid());
        Header header = createHeader(xidLogEvent.getLogHeader(), "", "", null);
        return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
    }

    public TableMetaData parseRowsEventForTableMeta(RowsLogEvent rowsLogEvent)
    {
        TableMapLogEvent rowsLogEventTable = rowsLogEvent.getTableMapLogEvent();
        if (rowsLogEventTable == null)
        {
            // tableId对应的记录不存在
            throw new TableIdNotFoundException("not found tableId:" + rowsLogEvent.getTableId());
        }

        boolean isHeartBeat = isAliSQLHeartBeat(rowsLogEventTable.getDbName(), rowsLogEventTable.getTableName());
        boolean isRDSHeartBeat = tableMetaDataCache.isOnRDS() && isRDSHeartBeat(rowsLogEventTable.getDbName(), rowsLogEventTable.getTableName());

        String fullname = rowsLogEventTable.getDbName() + "." + rowsLogEventTable.getTableName();
        // check name eventFilter
        if (nameFilter != null && !nameFilter.filter(fullname))
        {
            return null;
        }
        if (nameBlackFilter != null && nameBlackFilter.filter(fullname))
        {
            return null;
        }

        // if (isHeartBeat || isRDSHeartBeat) {
        // // 忽略rds模式的mysql.ha_health_check心跳数据
        // return null;
        // }
        TableMetaData tableMetaData = null;
        if (isRDSHeartBeat)
        {
            // 处理rds模式的mysql.ha_health_check心跳数据
            // 主要RDS的心跳表基本无权限,需要mock一个tableMeta
            FieldMetaData idMeta = new FieldMetaData("id", "bigint(20)", true, "0", false);
            FieldMetaData typeMeta = new FieldMetaData("type", "char(1)", false, "0", true);
            tableMetaData = new TableMetaData(rowsLogEventTable.getDbName(), rowsLogEventTable.getTableName(), Arrays.asList(idMeta, typeMeta));
        } else if (isHeartBeat)
        {
            // 处理alisql模式的test.heartbeat心跳数据
            // 心跳表基本无权限,需要mock一个tableMeta
            FieldMetaData idMeta = new FieldMetaData("id", "smallint(6)", false, null, true);
            FieldMetaData typeMeta = new FieldMetaData("ts", "int(11)", true, null, false);
            tableMetaData = new TableMetaData(rowsLogEventTable.getDbName(), rowsLogEventTable.getTableName(), Arrays.asList(idMeta, typeMeta));
        }

        EntryPosition position = createPosition(rowsLogEvent.getLogHeader());
        if (tableMetaDataCache != null && tableMetaData == null)
        {// 入错存在table metadata
            tableMetaData = getTableMeta(rowsLogEventTable.getDbName(), rowsLogEventTable.getTableName(), true, position);
            if (tableMetaData == null)
            {
                if (!filterTableError)
                {
                    throw new BinlogParseException("not found [" + fullname + "] in db , pls check!");
                }
            }
        }

        return tableMetaData;
    }

    public Entry parseRowsEvent(RowsLogEvent rowsLogEvent)
    {
        return parseRowsEvent(rowsLogEvent, null);
    }

    public Entry parseRowsEvent(RowsLogEvent rowsLogEvent, TableMetaData tableMetaData)
    {
        if (filterRows)
        {
            return null;
        }
        try
        {
            if (tableMetaData == null)
            { // 如果没有外部指定
                tableMetaData = parseRowsEventForTableMeta(rowsLogEvent);
            }

            if (tableMetaData == null)
            {
                // 拿不到表结构,执行忽略
                return null;
            }

            EventType eventType = null;
            int type = rowsLogEvent.getLogHeader().getType();
            if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type)
            {
                eventType = EventType.INSERT;
            } else if (LogEvent.UPDATE_ROWS_EVENT_V1 == type || LogEvent.UPDATE_ROWS_EVENT == type)
            {
                eventType = EventType.UPDATE;
            } else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type)
            {
                eventType = EventType.DELETE;
            } else
            {
                throw new BinlogParseException("unsupport event type :" + rowsLogEvent.getLogHeader().getType());
            }

            RowChange.Builder rowChangeBuider = RowChange.newBuilder();
            rowChangeBuider.setTableId(rowsLogEvent.getTableId());
            rowChangeBuider.setIsDdl(false);

            rowChangeBuider.setEventType(eventType);
            RowsLogBuffer buffer = rowsLogEvent.getRowsBuf(charset.name());
            BitSet columns = rowsLogEvent.getColumns();
            BitSet changeColumns = rowsLogEvent.getChangeColumns();

            boolean tableError = false;
            int rowsCount = 0;
            while (buffer.nextOneRow(columns))
            {
                // 处理row记录
                RowData.Builder rowDataBuilder = RowData.newBuilder();
                if (EventType.INSERT == eventType)
                {
                    // insert的记录放在before字段中
                    tableError |= parseOneRow(rowDataBuilder, rowsLogEvent, buffer, columns, true, tableMetaData);
                } else if (EventType.DELETE == eventType)
                {
                    // delete的记录放在before字段中
                    tableError |= parseOneRow(rowDataBuilder, rowsLogEvent, buffer, columns, false, tableMetaData);
                } else
                {
                    // update需要处理before/after
                    tableError |= parseOneRow(rowDataBuilder, rowsLogEvent, buffer, columns, false, tableMetaData);
                    if (!buffer.nextOneRow(changeColumns))
                    {
                        rowChangeBuider.addRowDatas(rowDataBuilder.build());
                        break;
                    }

                    tableError |= parseOneRow(rowDataBuilder, rowsLogEvent, buffer, changeColumns, true, tableMetaData);
                }

                rowsCount++;
                rowChangeBuider.addRowDatas(rowDataBuilder.build());
            }
            TableMapLogEvent tableMapLogEvent = rowsLogEvent.getTableMapLogEvent();
            Header header = createHeader(rowsLogEvent.getLogHeader(),
                    tableMapLogEvent.getDbName(),
                    tableMapLogEvent.getTableName(),
                    eventType,
                    rowsCount);

            RowChange rowChange = rowChangeBuider.build();
            if (tableError)
            {
                Entry entry = createEntry(header, EntryType.ROWDATA, ByteString.EMPTY);
                logger.warn("table parser error : {}storeValue: {}", entry.toString(), rowChange.toString());
                return null;
            } else
            {
                Entry entry = createEntry(header, EntryType.ROWDATA, rowChange.toByteString());
                return entry;
            }
        } catch (Exception e)
        {
            throw new BinlogParseException("parse row data failed.", e);
        }
    }

    private EntryPosition createPosition(LogHeader logHeader)
    {
        return new EntryPosition(logHeader.getLogFileName(),
                logHeader.getLogPosition(),
                logHeader.getWhen() * 1000L,
                logHeader.getServerId()); // 记录到秒
    }

    private boolean parseOneRow(RowData.Builder rowDataBuilder, RowsLogEvent event, RowsLogBuffer rowsLogBuffer, BitSet cols,
                                boolean isAfter, TableMetaData tableMetaData) throws UnsupportedEncodingException
    {
        int columnCnt = event.getTableMapLogEvent().getColumnCnt();
        ColumnInfo[] columnInfos = event.getTableMapLogEvent().getColumnInfo();

        boolean tableError = false;
        // check table fileds count，只能处理加字段
        boolean existRDSNoPrimaryKey = false;
        if (tableMetaData != null && columnInfos.length > tableMetaData.getFieldMetaDataList().size())
        {
            if (tableMetaDataCache.isOnRDS())
            {
                // 特殊处理下RDS的场景
                List<FieldMetaData> primaryKeys = tableMetaData.getPrimaryKeyFieldMetaDataList();
                if (primaryKeys == null || primaryKeys.isEmpty())
                {
                    if (columnInfos.length == tableMetaData.getFieldMetaDataList().size() + 1
                            && columnInfos[columnInfos.length - 1].type == LogEvent.MYSQL_TYPE_LONGLONG)
                    {
                        existRDSNoPrimaryKey = true;
                    }
                }
            }

            EntryPosition position = createPosition(event.getLogHeader());
            if (!existRDSNoPrimaryKey)
            {
                // online ddl增加字段操作步骤：
                // 1. 新增一张临时表，将需要做ddl表的数据全量导入
                // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表
                // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作
                // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作
                // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作
                tableMetaData = getTableMeta(event.getTableMapLogEvent().getDbName(), event.getTableMapLogEvent().getTableName(), false, position);// 强制重新获取一次
                if (tableMetaData == null)
                {
                    tableError = true;
                    if (!filterTableError)
                    {
                        throw new BinlogParseException("not found [" + event.getTableMapLogEvent().getDbName() + "."
                                + event.getTableMapLogEvent().getTableName() + "] in db , pls check!");
                    }
                }

                // 在做一次判断
                if (tableMetaData != null && columnInfos.length > tableMetaData.getFieldMetaDataList().size())
                {
                    tableError = true;
                    if (!filterTableError)
                    {
                        throw new BinlogParseException("column size is not match for table:" + tableMetaData.getFullName()
                                + "," + columnInfos.length + " vs " + tableMetaData.getFieldMetaDataList().size());
                    }
                }
            } else
            {
                logger.warn("[" + event.getTableMapLogEvent().getDbName() + "." + event.getTableMapLogEvent().getTableName()
                        + "] is no primary key , skip alibaba_rds_row_id column");
            }
        }

        for (int i = 0; i < columnCnt; i++)
        {
            ColumnInfo columnInfo1 = columnInfos[i];
            // mysql 5.6开始支持nolob/mininal类型,并不一定记录所有的列,需要进行判断
            if (!cols.get(i))
            {
                continue;
            }

            if (existRDSNoPrimaryKey && i == columnCnt - 1 && columnInfo1.type == LogEvent.MYSQL_TYPE_LONGLONG)
            {
                // 不解析最后一列
                rowsLogBuffer.nextValue(columnInfo1.type, columnInfo1.meta, false);
                continue;
            }

            Column.Builder columnBuilder = Column.newBuilder();

            FieldMetaData fieldMetaData = null;
            if (tableMetaData != null && !tableError)
            {
                // 处理file metadata
                fieldMetaData = tableMetaData.getFieldMetaDataList().get(i);
                columnBuilder.setName(fieldMetaData.getColumnName());
                columnBuilder.setIsKey(fieldMetaData.isPrimaryKey());
                // 增加mysql type类型,issue 73
                columnBuilder.setMysqlType(fieldMetaData.getColumnType());
            }
            columnBuilder.setIndex(i);
            columnBuilder.setIsNull(false);

            // fixed issue
            // https://github.com/alibaba/canal/issues/66，特殊处理binary/varbinary，不能做编码处理
            boolean isBinary = false;
            boolean isSingleBit = false;
            if (fieldMetaData != null)
            {
                if (StringUtils.containsIgnoreCase(fieldMetaData.getColumnType(), "VARBINARY"))
                {
                    isBinary = true;
                } else if (StringUtils.containsIgnoreCase(fieldMetaData.getColumnType(), "BINARY"))
                {
                    isBinary = true;
                } else if (StringUtils.containsIgnoreCase(fieldMetaData.getColumnType(), "TINYINT(1)"))
                {
                    isSingleBit = true;
                }
            }
            rowsLogBuffer.nextValue(columnInfo1.type, columnInfo1.meta, isBinary);
            if (existRDSNoPrimaryKey && i == columnCnt - 1 && columnInfo1.type == LogEvent.MYSQL_TYPE_LONGLONG)
            {
                // 不解析最后一列
                continue;
            }

            int javaType = rowsLogBuffer.getJavaType();
            if (isSingleBit && javaType == Types.TINYINT)
            {
                javaType = Types.BIT;
            }
            if (rowsLogBuffer.isNull())
            {
                columnBuilder.setIsNull(true);
            } else
            {
                final Serializable value = rowsLogBuffer.getValue();
                // 处理各种类型
                switch (javaType)
                {
                    case Types.INTEGER:
                    case Types.TINYINT:
                    case Types.SMALLINT:
                    case Types.BIGINT:
                        // 处理unsigned类型
                        Number number = (Number) value;
                        if (fieldMetaData != null && fieldMetaData.isUnsigned() && number.longValue() < 0)
                        {
                            switch (rowsLogBuffer.getLength())
                            {
                                case 1: /* MYSQL_TYPE_TINY */
                                    columnBuilder.setValue(String.valueOf(Integer.valueOf(TINYINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.SMALLINT; // 往上加一个量级
                                    break;

                                case 2: /* MYSQL_TYPE_SHORT */
                                    columnBuilder.setValue(String.valueOf(Integer.valueOf(SMALLINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.INTEGER; // 往上加一个量级
                                    break;

                                case 3: /* MYSQL_TYPE_INT24 */
                                    columnBuilder.setValue(String.valueOf(Integer.valueOf(MEDIUMINT_MAX_VALUE
                                            + number.intValue())));
                                    javaType = Types.INTEGER; // 往上加一个量级
                                    break;

                                case 4: /* MYSQL_TYPE_LONG */
                                    columnBuilder.setValue(String.valueOf(Long.valueOf(INTEGER_MAX_VALUE
                                            + number.longValue())));
                                    javaType = Types.BIGINT; // 往上加一个量级
                                    break;

                                case 8: /* MYSQL_TYPE_LONGLONG */
                                    columnBuilder.setValue(BIGINT_MAX_VALUE.add(BigInteger.valueOf(number.longValue()))
                                            .toString());
                                    javaType = Types.DECIMAL; // 往上加一个量级，避免执行出错
                                    break;
                            }
                        } else
                        {
                            // 对象为number类型，直接valueof即可
                            columnBuilder.setValue(String.valueOf(value));
                        }
                        break;
                    case Types.REAL: // float
                    case Types.DOUBLE: // double
                        // 对象为number类型，直接valueof即可
                        columnBuilder.setValue(String.valueOf(value));
                        break;
                    case Types.BIT:// bit
                        // 对象为number类型
                        columnBuilder.setValue(String.valueOf(value));
                        break;
                    case Types.DECIMAL:
                        columnBuilder.setValue(((BigDecimal) value).toPlainString());
                        break;
                    case Types.TIMESTAMP:
                        // 修复时间边界值
                        // String v = value.toString();
                        // v = v.substring(0, v.length() - 2);
                        // columnBuilder.setValue(v);
                        // break;
                    case Types.TIME:
                    case Types.DATE:
                        // 需要处理year
                        columnBuilder.setValue(value.toString());
                        break;
                    case Types.BINARY:
                    case Types.VARBINARY:
                    case Types.LONGVARBINARY:
                        // fixed text encoding
                        // https://github.com/AlibabaTech/canal/issues/18
                        // mysql binlog中blob/text都处理为blob类型，需要反查table
                        // metadata，按编码解析text
                        if (fieldMetaData != null && isText(fieldMetaData.getColumnType()))
                        {
                            columnBuilder.setValue(new String((byte[]) value, charset));
                            javaType = Types.CLOB;
                        } else
                        {
                            // byte数组，直接使用iso-8859-1保留对应编码，浪费内存
                            columnBuilder.setValue(new String((byte[]) value, ISO_8859_1));
                            // columnBuilder.setValueBytes(ByteString.copyFrom((byte[])
                            // value));
                            javaType = Types.BLOB;
                        }
                        break;
                    case Types.CHAR:
                    case Types.VARCHAR:
                        columnBuilder.setValue(value.toString());
                        break;
                    default:
                        columnBuilder.setValue(value.toString());
                }

            }

            columnBuilder.setSqlType(javaType);
            // 设置是否update的标记位
            columnBuilder.setUpdated(isAfter
                    && isUpdate(rowDataBuilder.getBeforeColumnsList(),
                    columnBuilder.getIsNull() ? null : columnBuilder.getValue(),
                    i));
            if (isAfter)
            {
                rowDataBuilder.addAfterColumns(columnBuilder.build());
            } else
            {
                rowDataBuilder.addBeforeColumns(columnBuilder.build());
            }
        }

        return tableError;

    }

    private Entry buildQueryEntry(String queryString, LogHeader logHeader, String tableName)
    {
        Header header = createHeader(logHeader, "", tableName, EventType.QUERY);
        RowChange.Builder rowChangeBuider = RowChange.newBuilder();
        rowChangeBuider.setSql(queryString);
        rowChangeBuider.setEventType(EventType.QUERY);
        return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
    }

    private Entry buildQueryEntry(String queryString, LogHeader logHeader)
    {
        Header header = createHeader(logHeader, "", "", EventType.QUERY);
        RowChange.Builder rowChangeBuider = RowChange.newBuilder();
        rowChangeBuider.setSql(queryString);
        rowChangeBuider.setEventType(EventType.QUERY);
        return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
    }

    private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType)
    {
        return createHeader(logHeader, schemaName, tableName, eventType, -1);
    }

    private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType,
                                Integer rowsCount)
    {
        // header会做信息冗余,方便以后做检索或者过滤
        Header.Builder headerBuilder = Header.newBuilder();
        headerBuilder.setVersion(version);
        headerBuilder.setLogfileName(logHeader.getLogFileName());
        headerBuilder.setLogfileOffset(logHeader.getLogPosition() - logHeader.getEventLength());
        headerBuilder.setServerId(logHeader.getServerId());
        headerBuilder.setServerenCode(UTF_8);// 经过java输出后所有的编码为unicode
        headerBuilder.setExecuteTime(logHeader.getWhen() * 1000L);
        headerBuilder.setSourceType(Type.MYSQL);
        if (eventType != null)
        {
            headerBuilder.setEventType(eventType);
        }
        if (schemaName != null)
        {
            headerBuilder.setSchemaName(schemaName);
        }
        if (tableName != null)
        {
            headerBuilder.setTableName(tableName);
        }
        headerBuilder.setEventLength(logHeader.getEventLength());
        // enable gtid index$originIndex
        if (StringUtils.isNotEmpty(logHeader.getGtidSetStr()))
        {
            headerBuilder.setGtid(logHeader.getGtidSetStr());
        }
        // add current gtid
        if (StringUtils.isNotEmpty(logHeader.getCurrentGtid()))
        {
            Pair pair = createSpecialPair("curtGtid", logHeader.getCurrentGtid());
            headerBuilder.addProps(pair);
        }
        // add current gtid sequence no
        if (StringUtils.isNotEmpty(logHeader.getCurrentGtidSn()))
        {
            Pair pair = createSpecialPair("curtGtidSn", logHeader.getCurrentGtidSn());
            headerBuilder.addProps(pair);
        }

        // add current gtid last committed
        if (StringUtils.isNotEmpty(logHeader.getCurrentGtidLastCommit()))
        {
            Pair pair = createSpecialPair("curtGtidLct", logHeader.getCurrentGtidLastCommit());
            headerBuilder.addProps(pair);
        }

        // add rowsCount suppport
        if (rowsCount > 0)
        {
            Pair pair = createSpecialPair("rowsCount", String.valueOf(rowsCount));
            headerBuilder.addProps(pair);
        }
        return headerBuilder.build();
    }

    private boolean isUpdate(List<Column> bfColumns, String newValue, int index)
    {
        if (bfColumns == null)
        {
            throw new BinlogParseException("ERROR ## the bfColumns is null");
        }

        if (index < 0)
        {
            return false;
        }

        for (Column column : bfColumns)
        {
            if (column.getIndex() == index)
            {// 比较before / after的column index$originIndex
                if (column.getIsNull() && newValue == null)
                {
                    // 如果全是null
                    return false;
                } else if (newValue != null && (!column.getIsNull() && column.getValue().equals(newValue)))
                {
                    // fixed issue #135, old column is Null
                    // 如果不为null，并且相等
                    return false;
                }
            }
        }

        // 比如nolob/minial模式下,可能找不到before记录,认为是有变化
        return true;
    }

    private TableMetaData getTableMeta(String dbName, String tbName, boolean useCache, EntryPosition position)
    {
        try
        {
            return tableMetaDataCache.getTableMeta(dbName, tbName, useCache, position);
        } catch (Throwable e)
        {
            String message = ExceptionUtils.getRootCauseMessage(e);
            if (filterTableError)
            {
                if (StringUtils.contains(message, "errorNumber=1146") && StringUtils.contains(message, "doesn't exist"))
                {
                    return null;
                } else if (StringUtils.contains(message, "errorNumber=1142")
                        && StringUtils.contains(message, "command denied"))
                {
                    return null;
                }
            }

            throw new BinlogParseException(e);
        }
    }

    private boolean isText(String columnType)
    {
        return "LONGTEXT".equalsIgnoreCase(columnType) || "MEDIUMTEXT".equalsIgnoreCase(columnType)
                || "TEXT".equalsIgnoreCase(columnType) || "TINYTEXT".equalsIgnoreCase(columnType);
    }

    private boolean isAliSQLHeartBeat(String schema, String table)
    {
        return "test".equalsIgnoreCase(schema) && "heartbeat".equalsIgnoreCase(table);
    }

    private boolean isRDSHeartBeat(String schema, String table)
    {
        return "mysql".equalsIgnoreCase(schema) && "ha_health_check".equalsIgnoreCase(table);
    }

    public void setCharset(Charset charset)
    {
        this.charset = charset;
    }

    public void setNameFilter(AviaterRegexFilter nameFilter)
    {
        this.nameFilter = nameFilter;
    }

    public void setNameBlackFilter(AviaterRegexFilter nameBlackFilter)
    {
        this.nameBlackFilter = nameBlackFilter;
    }

    public void setTableMetaDataCache(TableMetaDataCache tableMetaDataCache)
    {
        this.tableMetaDataCache = tableMetaDataCache;
    }

    public void setFilterQueryDcl(boolean filterQueryDcl)
    {
        this.filterQueryDcl = filterQueryDcl;
    }

    public void setFilterQueryDml(boolean filterQueryDml)
    {
        this.filterQueryDml = filterQueryDml;
    }

    public void setFilterQueryDdl(boolean filterQueryDdl)
    {
        this.filterQueryDdl = filterQueryDdl;
    }

    public void setFilterTableError(boolean filterTableError)
    {
        this.filterTableError = filterTableError;
    }

    public void setFilterRows(boolean filterRows)
    {
        this.filterRows = filterRows;
    }
}
